package ds_industry_2025.industry.gy_10.T3

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

/*
    2、编写scala代码，使用Spark根据hudi_gy_dwd层的fact_machine_data表统计出每日每台设备，状态为“运行”的时长（
    若运行无结束时间，则需根据时间判断这个设备的运行状态的下一个状态是哪条数据，将下一个状态的数据的时间置为这个设备运行状态的结
    束时间,如果设备数据的运行状态不存在下一个状态，则该设备这个阶段数据的运行状态不参与计算，即该设备的这个阶段数据的运行状态时长
    按0计算），将结果数据写入hudi_gy_dws层的表machine_data_total_time中，然后使用spark-sql的cli根据machine_id降序
    和machine_record_date升序排序查询前5条数据，将SQL语句复制粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任
    务序号下，将执行结果截图粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下；
 */
object t5 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t4")
      .config("hive.exec.dynamic.partition.mode", "nonstrict")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .config("spark.sql.legacy.parquet.datetimeRebaseModeInRead", "LEGACY")
      .enableHiveSupport()
      .getOrCreate()

    val hdfs="hdfs://192.168.40.110:9000/user/hive/warehouse/hudi_gy_dwd.db/fact_machine_data"

    val data = spark.read.format("hudi").load(hdfs)

    // todo 定义一个窗口排序规则
    val window=Window
      .partitionBy(col("machineid"),to_date(col("machinerecorddate")))
      .orderBy("machinerecorddate")

    //  todo 拿到结果
    val result = data.select("machineid", "machinerecordstate", "machinerecorddate")
      .withColumn(
        "end_time",
        lead("machinerecorddate", 1).over(window)
      )
      .filter(col("end_time").isNotNull)
      .filter(col("machinerecordstate") === "运行")
      .withColumn(
        "diff",
        unix_timestamp(col("end_time")) - unix_timestamp(col("machinerecorddate"))
      )
      .withColumn("machine_record_date", to_date(col("machinerecorddate")))
      .groupBy("machineid", "machine_record_date")
      .agg(
        sum("diff").as("total_amount")
      )
      .withColumnRenamed("machineid", "machine_id")
      .distinct()

    result.show

    result.write.mode("overwrite").format("hive")
      .saveAsTable("hudi_gy_dws.machine_data_total_time")


    spark.close()

  }

}
